package com.atguigu.day06;

import com.atguigu.util.ProductViewCountPerWindow;
import com.atguigu.util.UserBehavior;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;

public class Example2 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        var source = FileSource
                .forRecordStreamFormat(
                        // 按行消费
                        new TextLineInputFormat(),
                        new Path("/home/yuantuzhi/flinktutorial0926/src/main/resources/UserBehavior.csv")
                )
                .build();

        env
                .fromSource(source, WatermarkStrategy.noWatermarks(), "user-behavior")
                .flatMap(new StringToUserBehavior())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner((element, recordTimestamp) -> element.ts)
                )
                .keyBy(r -> r.productId)
                .process(new MySlidingEventTimeWindow(60 * 60 * 1000L, 5 * 60 * 1000L))
                .keyBy(r -> r.windowEndTime)
                .process(new TopN(3))
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class StringToUserBehavior implements FlatMapFunction<String, UserBehavior> {
        @Override
        public void flatMap(String in, Collector<UserBehavior> out) throws Exception {
            var fields = in.split(",");
            var userBehavior = new UserBehavior(
                    fields[0], fields[1], fields[2], fields[3],
                    Long.parseLong(fields[4]) * 1000L
            );

            if (userBehavior.type.equals("pv"))
                out.collect(userBehavior);
        }
    }

    public static class MySlidingEventTimeWindow extends KeyedProcessFunction<String, UserBehavior, ProductViewCountPerWindow> {
        private final long windowSize;
        private final long windowSlide;

        public MySlidingEventTimeWindow(long windowSize, long windowSlide) {
            this.windowSize = windowSize;
            this.windowSlide = windowSlide;
        }

        private MapState<Tuple2<Long, Long>, Long> mapState;

        @Override
        public void open(Configuration parameters) throws Exception {
            mapState = getRuntimeContext().getMapState(
                    new MapStateDescriptor<Tuple2<Long, Long>, Long>(
                            "windowInfo-accumulator",
                            Types.TUPLE(Types.LONG, Types.LONG),
                            Types.LONG
                    )
            );
        }

        @Override
        public void processElement(UserBehavior in, Context ctx, Collector<ProductViewCountPerWindow> out) throws Exception {
            var firstWindowStartTime = in.ts - in.ts % windowSlide;
            var windowInfoArray = new ArrayList<Tuple2<Long, Long>>();
            for (var start = firstWindowStartTime; start > in.ts - windowSize; start = start - windowSlide) {
                windowInfoArray.add(Tuple2.of(start, start + windowSize));
            }

            for (var windowInfo : windowInfoArray) {
                if (!mapState.contains(windowInfo)) {
                    mapState.put(windowInfo, 1L);
                } else {
                    mapState.put(windowInfo, mapState.get(windowInfo) + 1L);
                }

                ctx.timerService().registerEventTimeTimer(windowInfo.f1 - 1L);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<ProductViewCountPerWindow> out) throws Exception {
            long windowEnd = timestamp + 1L;
            long windowStart = windowEnd - windowSize;
            var windowInfo = Tuple2.of(windowStart, windowEnd);
            var count = mapState.get(windowInfo);
            var productId = ctx.getCurrentKey();
            out.collect(new ProductViewCountPerWindow(productId, count, windowStart, windowEnd));

            mapState.remove(windowInfo);
        }
    }

    public static class TopN extends KeyedProcessFunction<Long, ProductViewCountPerWindow, String> {
        private final int n;

        public TopN(int n) {
            this.n = n;
        }

        private ListState<ProductViewCountPerWindow> listState;

        @Override
        public void open(Configuration parameters) throws Exception {
            listState = getRuntimeContext().getListState(
                    new ListStateDescriptor<ProductViewCountPerWindow>(
                            "list",
                            Types.POJO(ProductViewCountPerWindow.class)
                    )
            );
        }

        @Override
        public void processElement(ProductViewCountPerWindow productViewCountPerWindow, Context context, Collector<String> collector) throws Exception {
            listState.add(productViewCountPerWindow);

            context.timerService().registerEventTimeTimer(productViewCountPerWindow.windowEndTime + 1L);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            var arr = new ArrayList<ProductViewCountPerWindow>();

            for (var e : listState.get()) arr.add(e);

            listState.clear();

            arr.sort((o1, o2) -> (int) (o2.count - o1.count));

            var result = new StringBuilder();
            result.append("=============窗口结束时间：").append(new Timestamp(timestamp - 1L)).append("==============\n");
            for (int i = 0; i < n; i++) {
                var tmp = arr.get(i);
                result.append("No.").append(i+1).append(",ID:").append(tmp.productId).append(",pv:").append(tmp.count).append("\n");
            }
            result.append("=============窗口结束时间：").append(new Timestamp(timestamp - 1L)).append("==============\n");
            out.collect(result.toString());
        }
    }
}
